package top.wshape1.rpc.consumer;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import top.wshape1.rpc.common.constant.RoleType;
import top.wshape1.rpc.common.entity.ServiceMeta;
import top.wshape1.rpc.consumer.handler.ConsumerChannelInitializer;
import top.wshape1.rpc.registry.RpcRegistry;
import top.wshape1.rpc.registry.RpcRegistryFactory;

import java.lang.reflect.Proxy;
import java.net.ConnectException;

/**
 * @author Wshape1
 * @version 1.0
 * @description Rpc调用方
 * @since 2024-02-02
 */

@Slf4j
public class RpcConsumer {

    private Bootstrap bootstrap;
    private NioEventLoopGroup eventLoopGroup;

    private ConsumerSession session;

    public RpcConsumer() {
        this.bootstrap = new Bootstrap();
        this.eventLoopGroup = new NioEventLoopGroup();
        this.session = new ConsumerSession();

        bootstrap.group(eventLoopGroup);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ConsumerChannelInitializer());
    }

    /**
     * 与服务提供方建立连接
     *
     * @param inetHost ip
     * @param inetPort port
     * @return Netty Channel
     */
    public Channel connect(String inetHost, int inetPort) throws ConnectException {
        if (session != null) {
            // 先从session中获取channel
            Channel ch = session.getChannel(inetHost, inetPort);
            if (ch != null) {
                log.debug("RpcConsumer已连接过服务({}：{})", inetHost, inetPort);
                return ch;
            }
        }

        log.info("Connecting to provider '{}:{}'.", inetHost, inetPort);
        try {
            Channel channel = bootstrap.connect(inetHost, inetPort).sync().channel();
            channel.closeFuture().addListener(future -> {
                log.info("Disconnect from provider '{}:{}'.", inetHost, inetPort);
                if (session != null) {
                    // 移除session
                    session.removeChannel(inetHost, inetPort);
                }
            });
            if (session != null) {
                // 添加到session，方便下次使用
                session.setChannel(inetHost, inetPort, channel);
            }

            return channel;
        } catch (Exception e) {
            log.error("An error occurred during the connection to provider '{}:{}'. ERROR: {}", inetHost, inetPort, e.getMessage());
            throw new ConnectException(e.getMessage());
        }
    }

    /**
     * 关闭RpcConsumer
     */
    public void stop() {
        session.clearAndClose();
        session = null;
        eventLoopGroup.shutdownGracefully();
        eventLoopGroup = null;
        bootstrap = null;
    }

    /**
     * 获取远程服务的代理对象
     *
     * @param serviceInterfaceClass 远程服务的接口类
     * @param <T>                   T
     * @return T
     */
    @SuppressWarnings("unchecked")
    public <T> T getProxyService(Class<T> serviceInterfaceClass) {
        RpcRegistry registry = RpcRegistryFactory.getRegistry();
        String serviceName = serviceInterfaceClass.getName();
        // 注册消费者，同时订阅生产者
        registry.register(ServiceMeta.selfServiceMeta(serviceName, RoleType.consumer));


        // JDK动态代理
        Object o = Proxy.newProxyInstance(serviceInterfaceClass.getClassLoader(),
                new Class[]{serviceInterfaceClass}, new RpcProxyInvocationHandler(serviceInterfaceClass, this));

        return (T) o;
    }
}
